package org.wzq;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.Watcher;

/**
 * author:wzq
 * date:16-6-28
 * project:zk_acl
 * description: TODO
 */
public class MqApp {

    //    private static final String zkAddress = "104.236.154.96:2181,198.199.108.10:2181,159.203.221.126:2181";
    private static final String zkAddress = "localhost:2181";
    private static final String sysConfig = "/sysConfig";
    private static final String appContext = "rabbit-mq";
    private static final String typePath = sysConfig + "/" + appContext;
    private static final String mqRead = "mqRead:123456";

//    private String appContext;  // type
    private String appId;       // target
    private String watchPath;
    private ZkClient zkClient;

    public MqApp(String appId) {
        this.appId = appId;
        this.watchPath = typePath + "/" + appId;
        this.zkClient = new ZkClient(zkAddress);
        System.out.println("appId=" + appId);
        System.out.println("watchPath=" + this.watchPath);
    }

    // 权限控制
    // 创建 znode /config if need
    // 应用启动: 注册自己到zk
    // 监听该节点
    public static void main(String[] args) {
        MqApp app = new MqApp("mq-demo");
        app.start();
        // app.keepAlive();
    }

    public void keepAlive() {
        new Thread(new Runnable() {
            public void run() {
                while (true) {
                    try {
                        Thread.sleep(10000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }).start();
    }

    public void start() {
        // ZkClient zkClient = new ZkClient(zkAddress);
        // create config if need
        if (!zkClient.exists(sysConfig)) {
            zkClient.createPersistent(sysConfig);
        }
        if (!zkClient.exists(typePath)) {
            zkClient.createPersistent(typePath);
        }

        // add appId
        zkClient.delete(this.watchPath);
        zkClient.createPersistent(this.watchPath);

        // listen thisId
        zkClient.subscribeDataChanges(this.watchPath, new IZkDataListener() {
            public void handleDataChange(String s, Object o) throws Exception {
                System.out.println("handleDataChange, path=" + s + "Object=" + o.toString());
            }

            public void handleDataDeleted(String s) throws Exception {
                System.out.println("handleDataDeleted");
                System.out.println("path=" + s);
            }
        });

        zkClient.subscribeStateChanges(new IZkStateListener() {
            public void handleStateChanged(Watcher.Event.KeeperState keeperState) throws Exception {
                System.out.println("handleStateChanged");
                System.out.println("keeperState=" + keeperState);
            }

            public void handleNewSession() throws Exception {
                System.out.println("handleStateChanged");
            }

            public void handleSessionEstablishmentError(Throwable throwable) throws Exception {
                System.out.println("handleSessionEstablishmentError");
            }
        });


//        // test
//        String testValue = "abc";
//        zkClient.writeData(this.watchPath, testValue);
//        String ret = zkClient.readData(this.watchPath);
//        System.out.println(ret);

        int start = 100;
        while (true) {
            // test
            String testValue = start + "";
            zkClient.writeData(this.watchPath, testValue);
            // String ret = zkClient.readData(this.watchPath);
            // System.out.println(ret);
            start++;

            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
